package org.example;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;

import java.io.IOException;
import java.io.PrintStream;
import java.net.URI;

/**
 * @author zhl
 * @version 1.0
 * @ClassName
 * @date 2020/11/2 17:57
 */

class MyPathFilter implements PathFilter {
    String reg = null;

    MyPathFilter(String reg) {
        this.reg = reg;
    }

    @Override
    public boolean accept(Path path) {
        return !(path.toString().matches(reg));
    }
}

/**
 * 利用FSDataOutputStream和FSD爱他InputStream合并HDFS中的文件
 * */
public class FileMerge {
    Path inputPath = null;  // 待合并的文件的目录的路径
    Path outputPath = null; // 输出文件路径

    public FileMerge(String input, String output) {
        this.inputPath = new Path(input);
        this.outputPath = new Path(output);
    }

    public static void main(String[] args) throws IOException {
        FileMerge merge = new FileMerge("hdfs://192.168.136.132:9000/user/hadoop/input",
                "hdfs://192.168.136.132:9000/user/hadoop/merge1.txt");
        merge.doMerge();
    }

    public void doMerge() throws IOException {
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://192.168.136.132:9000");
        FileSystem fsSource = FileSystem.get(URI.create(inputPath.toString()), conf);
        FileSystem fsDst = FileSystem.get(URI.create(outputPath.toString()), conf);
        // 下面过滤掉目录中后缀为.xml的文件
        FileStatus[] sourceStatus = fsSource.listStatus(inputPath, new MyPathFilter(".*\\.xml"));
        FSDataOutputStream fsdOS = fsDst.create(outputPath);
        PrintStream ps = new PrintStream(System.out);
        // 下面分别读取过滤之后的每个文件的内容，并输出到同一个文件中
        for (FileStatus sta : sourceStatus) {
            // 下面打印后缀部位.abc的文件的路径、文件大小
            System.out.println("路径：" + sta.getPath() + " 文件大小:" + sta.getLen() + " 权限：" +
                    sta.getPermission() + " 内容：");
            FSDataInputStream fsdIS = fsSource.open(sta.getPath());

//            byte[] data = new byte[1024];
//            int read = -1;

//            while ((read = fsdIS.read(data)) > 0) {
//                ps.write(data, 0, read);
//                fsdOS.write(data, 0, read);
//            }
            fsdIS.close();
        }
        ps.close();
        fsdOS.close();
    }

}
